// 版权所有2009 Go作者。保留所有权利。
// 此源代码的使用受BSD样式
// 许可证的约束，该许可证可以在许可证文件中找到。

// 用于连接需要io的代码的管道适配器。读卡器
// 代码需要io。作家

package io

import (
	"errors"
	"sync"
)

// onceError是一个只存储一次错误的对象。
type onceError struct {
	sync.Mutex // 
	err        error
}

func (a *onceError) Store(err error) {
	a.Lock()
	defer a.Unlock()
	if a.err != nil {
		return
	}
	a.err = err
}
func (a *onceError) Load() error {
	a.Lock()
	defer a.Unlock()
	return a.err
}

// ErrClosedPipe是用于在封闭管道上执行读写操作的错误。
var ErrClosedPipe = errors.New("io: read/write on closed pipe")

// 管道是PipeReader和PipeWriter下面的共享管道结构。
type pipe struct {
	wrMu sync.Mutex // 序列化写入操作
	wrCh chan []byte
	rdCh chan int

	once sync.Once // 保护关闭完成
	done chan struct{}
	rerr onceError
	werr onceError
}

func (p *pipe) read(b []byte) (n int, err error) {
	select {
	case <-p.done:
		return 0, p.readCloseError()
	default:
	}

	select {
	case bw := <-p.wrCh:
		nr := copy(b, bw)
		p.rdCh <- nr
		return nr, nil
	case <-p.done:
		return 0, p.readCloseError()
	}
}

func (p *pipe) closeRead(err error) error {
	if err == nil {
		err = ErrClosedPipe
	}
	p.rerr.Store(err)
	p.once.Do(func() { close(p.done) })
	return nil
}

func (p *pipe) write(b []byte) (n int, err error) {
	select {
	case <-p.done:
		return 0, p.writeCloseError()
	default:
		p.wrMu.Lock()
		defer p.wrMu.Unlock()
	}

	for once := true; once || len(b) > 0; once = false {
		select {
		case p.wrCh <- b:
			nw := <-p.rdCh
			b = b[nw:]
			n += nw
		case <-p.done:
			return n, p.writeCloseError()
		}
	}
	return n, nil
}

func (p *pipe) closeWrite(err error) error {
	if err == nil {
		err = EOF
	}
	p.werr.Store(err)
	p.once.Do(func() { close(p.done) })
	return nil
}

// readCloseError被认为是管道类型的内部错误。
func (p *pipe) readCloseError() error {
	rerr := p.rerr.Load()
	if werr := p.werr.Load(); rerr == nil && werr != nil {
		return werr
	}
	return ErrClosedPipe
}

// writeCloseError被认为是管道类型的内部错误。
func (p *pipe) writeCloseError() error {
	werr := p.werr.Load()
	if rerr := p.rerr.Load(); werr == nil && rerr != nil {
		return rerr
	}
	return ErrClosedPipe
}

// 管道读取器是管道的读取部分。
type PipeReader struct {
	p *pipe
}

// Read实现标准的读取接口：
// 它从管道中读取数据，直到写入程序
// 到达或写入端关闭。
// 如果写入端因错误而关闭，则该错误为
// 返回为err；否则err就是EOF。
func (r *PipeReader) Read(data []byte) (n int, err error) {
	return r.p.read(data)
}

// 关闭阅读器；后续写入
// 写入一半管道将返回错误ErrClosedPipe。
func (r *PipeReader) Close() error {
	return r.CloseWithError(nil)
}

// CloseWitheror关闭阅读器；后续写入
// 到管道的写入部分将返回错误err。wen jian defg
func (r *PipeReader) CloseWithError(err error) error {
	return r.p.closeRead(err)
}

// 管道编写器是管道的一半。
type PipeWriter struct {
	p *pipe
}

// Write实现了标准的写入接口：
// 它将数据写入管道，直到一个或多个读卡器
// 消耗了所有数据或读取端关闭。
// 如果读取端因错误而关闭，则该err为
// 返回为err；否则err就是ErrClosedPipe。
func (w *PipeWriter) Write(data []byte) (n int, err error) {
	return w.p.write(data)
}

// 关闭编写器；从
// 读取一半管道将不返回字节和EOF。
func (w *PipeWriter) Close() error {
	return w.CloseWithError(nil)
}

// CloseWitheror关闭了作者；从
// 读取一半管道的后续读取将不返回任何字节，如果err为零，则返回错误err、
// 或EOF。
// 
// CloseWithError从不覆盖上一个错误（如果它存在），并且始终返回零。
func (w *PipeWriter) CloseWithError(err error) error {
	return w.p.closeWrite(err)
}

// 管道创建一个同步内存管道。
// 可用于连接需要io的代码。读卡器
// 代码需要io。作家
// 
// 管道上的读写操作是一对一匹配的
// 除非一次写入需要多次读取。
// 也就是说，每次对PipeWriter的写入都会阻塞，直到满足
// 从完全使用
// 写入数据的PipeReader中读取一次或多次。
// 数据直接从写入复制到相应的
// 读取（或读取）；没有内部缓冲。
// 
// 可以安全地调用读写并行或关闭。
// 读取的并行调用和写入的并行调用也是安全的：
// 各个调用将按顺序被选通。
func Pipe() (*PipeReader, *PipeWriter) {
	p := &pipe{
		wrCh: make(chan []byte),
		rdCh: make(chan int),
		done: make(chan struct{}),
	}
	return &PipeReader{p}, &PipeWriter{p}
}
